/**
 * @author: yasen
 * @time: 2022/10/10 11:08
 * @email: yasen@yasenstudio.com
 */
package org.example;

import com.yasenstudio.flow.tools.SimpleKafkaConsumerBuilder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;

public class KafkaMysqlRecorderApp {
    private static Logger logger = LoggerFactory.getLogger(KafkaMysqlRecorderApp.class);
    public static void main( String[] args ) throws IOException {
        try (KafkaConsumer<String, String> kafkaConsumer = new SimpleKafkaConsumerBuilder("192.168.36.139:9092")
                .build()) {
            kafkaConsumer.subscribe(Collections.singletonList("ar_previous_data_test"));
            ConsumerRecordDAO recorder = new SimpleRecordDAOProxy();
            while (true) {
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    recorder.create(consumerRecord. key(), consumerRecord.value());
                    logger.info(consumerRecord.value());
                }
            }
        } catch (SQLException | FileNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
}